package regular;
import java.io.IOException;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.Queue;

import org.apache.hadoop.filecache.DistributedCache;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.BooleanWritable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.Reducer;

public class EvalSecondReducer extends
		Reducer<IntWritable, NodeWritable, NullWritable, BooleanWritable> {

	public void reduce(IntWritable key, Iterable<NodeWritable> values,
			Context context) throws IOException, InterruptedException {
		boolean answer = false;

		Iterator<NodeWritable> itr = values.iterator();
		NodeWritable data = new NodeWritable();
		HashMap<Integer,Node> listNode = new HashMap<Integer,Node>();
		while (itr.hasNext()) {
			data = itr.next();
			listNode.putAll(data.listNode);
		}

//		for(Node n: listNode.values())
//		{
//			System.out.print(n.id);
//			for(int e: n.neighbors)
//				System.out.print("\t"+e);
//			System.out.println();
//		}
		
		// get query
		Path[] fileCaches = DistributedCache.getLocalCacheFiles(context
				.getConfiguration());
		GraphQuery gq = Ultility.getAutomaton(fileCaches[0]);

		answer = evalDG(listNode, gq);

		context.write(NullWritable.get(), new BooleanWritable(answer));

		int countEdge = 0;
		for (Node n : listNode.values())
			countEdge += n.neighbors.size();

		System.out.println("Size of dependency graph: " + countEdge);
	}

	// A BFS based function to check whether s can reach t
	public static boolean isReachable(HashMap<Integer,Node> Vd, Node s, Node t) {
		if (s.equals(t))
			return true;

		// Create a queue for BFS
		Queue<Node> queue = new LinkedList<Node>();
		s.visit = true;
		queue.add(s);

		while (!queue.isEmpty()) {
			Node v = queue.poll();
			
			for (int e : v.neighbors) {
				// If this neighbor is the target vertex, then return true
				if (t.equals(new Node(e)))
					return true;
				// Else, continue to do BFS
				//int id = Vd.indexOf(new Node(e));
				Node tmp = Vd.get(e);
				if (tmp != null) {
					if (tmp.visit == false) {
						queue.add(tmp);
						Vd.get(e).visit = true;						
					}
				}
			}
		}
		return false;
	}

	public boolean evalDG(HashMap<Integer,Node> Vd, GraphQuery gq) {
		boolean answer = false;
		
		
		Node t = new Node(Integer.parseInt(gq.ut.stateName));

		Node startNode = Vd.get(Integer.parseInt(gq.us.stateName));
		
		if(startNode == null){			
			return false;
		}
		Node terminalNode = Vd.get(Integer.parseInt(gq.ut.stateName));
		if(terminalNode==null){
			Vd.put(t.id, t);	
			terminalNode = t;
		}		
		
		// BFS to get final answer
		answer = isReachable(Vd, startNode,terminalNode);
		return answer;
	}
}
